package com.atguigu.bigdata.spark.core.rdd.req;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple4;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/*
            // Q : actionRDD重复使用
        // Q : cogroup性能可能较低
 */
public class Spark02_Req1_HotCategoryTop10Analysis1_JAVA {
    public static void main(String[] args) {
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis");
        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        //读取原始日志数据

        JavaRDD<String> actionRDD = sc.textFile("datas/user_visit_action.txt");

        actionRDD.cache();

        // 2. 统计品类的点击数量：（品类ID，点击数量）
        JavaRDD<String> clickActionRDD_1 = actionRDD.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[6].equals("-1"));
            }
        });

        JavaPairRDD<String,Integer> clickActionRDD_2 = clickActionRDD_1.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                String[] temp = s.split("_");
                return new Tuple2<String, Integer>(temp[6], 1);
            }
        });

        JavaPairRDD<String,Integer>  clickActionRDD = clickActionRDD_2.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });

        // 3. 统计品类的下单数量：（品类ID，下单数量）

        JavaRDD<String> orderActionRDD_1 = actionRDD.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[8].equals("null"));
            }
        });
        JavaPairRDD<String,Integer> orderActionRDD_2 = orderActionRDD_1.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] temp = s.split("_");
                List<Tuple2<String, Integer>> list = new ArrayList<>();
                String[] temp1 = temp[8].split(",");
                for(String val : temp1) {
                    list.add(new Tuple2<String, Integer>(val ,1));
                }
                return list.iterator();
            }
        });

        JavaPairRDD<String, Integer> orderActionRDD = orderActionRDD_2.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        //// 4. 统计品类的支付数量：（品类ID，支付数量）
        JavaRDD<String> payActionRDD_1 = actionRDD.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[10].equals("null"));
            }
        });
        JavaPairRDD<String,Integer>  payActionRDD_2 = payActionRDD_1.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] temp = s.split("_");
                List<Tuple2<String, Integer>> list = new ArrayList<>();
                String[] temp1 = temp[10].split(",");
                for(String val : temp1) {
                    list.add(new Tuple2<String, Integer>(val ,1));
                }
                return list.iterator();
            }
        });
        JavaPairRDD<String, Integer> payActionRDD = payActionRDD_2.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });

        // (品类ID, 点击数量) => (品类ID, (点击数量, 0, 0))
        // (品类ID, 下单数量) => (品类ID, (0, 下单数量, 0))
        //                    => (品类ID, (点击数量, 下单数量, 0))
        // (品类ID, 支付数量) => (品类ID, (0, 0, 支付数量))
        //                    => (品类ID, (点击数量, 下单数量, 支付数量))
        // ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>>  click = clickActionRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple2<String, Tuple3<Integer, Integer, Integer>> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return new Tuple2<String, Tuple3<Integer, Integer, Integer>>(stringIntegerTuple2._1,
                        new Tuple3<Integer, Integer, Integer>(stringIntegerTuple2._2,0,0));
            }
        });

        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>>  order = orderActionRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple2<String, Tuple3<Integer, Integer, Integer>> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return new Tuple2<String, Tuple3<Integer, Integer, Integer>>(stringIntegerTuple2._1,
                        new Tuple3<Integer, Integer, Integer>(0,stringIntegerTuple2._2,0));
            }
        });

        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>>  pay = payActionRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple2<String, Tuple3<Integer, Integer, Integer>> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return new Tuple2<String, Tuple3<Integer, Integer, Integer>>(stringIntegerTuple2._1,
                        new Tuple3<Integer, Integer, Integer>(0,0,stringIntegerTuple2._2));
            }
        });
        // 5. 将品类进行排序，并且取前10名
        //    点击数量排序，下单数量排序，支付数量排序
        //    元组排序：先比较第一个，再比较第二个，再比较第三个，依此类推
        //    ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>> source = click.union(order).union(pay);

        JavaPairRDD<String, Tuple3<Integer, Integer, Integer>> analysis = source.reduceByKey(new Function2<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple3<Integer, Integer, Integer> call(Tuple3<Integer, Integer, Integer> p1,
                                                          Tuple3<Integer, Integer, Integer> p2) throws Exception {

                return new Tuple3<Integer, Integer, Integer>(p1._1()+ p2._1(), p1._2()+ p2._2(), p1._3() + p2._3());
            }
        });

        JavaRDD<Tuple4<String, Integer,Integer,Integer>> analysisMap = analysis.map(new Function<Tuple2<String, Tuple3<Integer, Integer, Integer>>, Tuple4<String, Integer, Integer, Integer>>() {
            @Override
            public Tuple4<String, Integer, Integer, Integer> call(Tuple2<String, Tuple3<Integer, Integer, Integer>> input) throws Exception {
                Tuple3<Integer, Integer, Integer> tuple3 = input._2;
                Tuple4<String, Integer, Integer, Integer> tuple4 = new Tuple4<>(input._1,tuple3._1(),tuple3._2(), tuple3._3());
                return tuple4;
            }
        });

        JavaRDD<Tuple4<String, Integer,Integer,Integer>> result = analysisMap.sortBy(new Function<Tuple4<String, Integer, Integer, Integer>, Object>() {
            @Override
            public Object call(Tuple4<String, Integer, Integer, Integer> tuple4) throws Exception {
                return tuple4._2();
            }
        },false,5);

        System.out.println(result.take(10).toString());

        sc.stop();

    }
}
